package test.simple;

import java.io.InputStream;
import java.net.URL;

import javax.jms.Connection;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;

// JMS local transactions
//http://fusesource.com/docs/esb/4.4.1/camel_tx/FMRTxnJMSConfig.html
//http://camel.apache.org/log.html
public class LaunchActiveMQLocalTransaction {

	public static void main(String[] args) throws Exception {

		
		org.apache.activemq.broker.BrokerService bs = new org.apache.activemq.broker.BrokerService();
		bs.setBrokerName("testBroker");
		bs.setPersistent(false);
		bs.setTransportConnectorURIs(new String[]{"tcp://localhost:61616"});
		bs.start();
		
		org.apache.activemq.ActiveMQConnectionFactory cf1 = new org.apache.activemq.ActiveMQConnectionFactory();
		cf1.setBrokerURL("tcp://localhost:61616");
		
		org.apache.activemq.ActiveMQConnectionFactory cf2 = new org.apache.activemq.ActiveMQConnectionFactory();
		cf2.setBrokerURL("tcp://localhost:61616");
		
		
		org.springframework.jms.connection.JmsTransactionManager txManager = new org.springframework.jms.connection.JmsTransactionManager();
		
		org.apache.camel.component.jms.JmsConfiguration jmsComp1 = new org.apache.camel.component.jms.JmsConfiguration();
		jmsComp1.setConnectionFactory(cf1);
		jmsComp1.setTransacted(false);
		jmsComp1.setConcurrentConsumers(1);
		
		org.apache.camel.component.jms.JmsConfiguration jmsComp2 = new org.apache.camel.component.jms.JmsConfiguration();
		jmsComp2.setConnectionFactory(cf2);
		jmsComp2.setTransacted(true);
		jmsComp2.setConcurrentConsumers(1);
		jmsComp2.setTransactionManager(txManager);
		jmsComp2.setCacheLevelName("CACHE_CONNECTION");
		
		txManager.setConnectionFactory(cf2);
		
		
		org.apache.activemq.camel.component.ActiveMQComponent comp1 = new org.apache.activemq.camel.component.ActiveMQComponent();
		comp1.setConfiguration(jmsComp1);
		
		org.apache.activemq.camel.component.ActiveMQComponent comp2 = new org.apache.activemq.camel.component.ActiveMQComponent();
		comp2.setConfiguration(jmsComp2);
		
		
		CamelContext context = new DefaultCamelContext();
		
		context.addComponent("activemq-non-tx", comp1);
		context.addComponent("activemq-local-tx", comp2);
	
		
		context.addRoutes(new RouteBuilder() {
			public void configure() {
				//from("activemq-local-tx:queue:inqueue").inOnly("activemq-local-tx:queue:outqueue").throwException(new Exception("Test Exception"));
				from("activemq-local-tx:queue:inqueue").to("log:qqqqqqqqqqqqqqqqqqqqqqqqqqqqq?level=INFO").inOnly("activemq-non-tx:queue:outqueue").throwException(new Exception("Test Exception"));
			}
		});
		


		context.start();
		Connection conn = cf1.createConnection();
		Session s = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
		 MessageProducer mp =s.createProducer(s.createQueue("inqueue"));
		 TextMessage tm = s.createTextMessage("Test message");
		mp.send(tm); 
		Thread.sleep(60000);
		context.stop();
		bs.stop();


	}


}
